package spark.work

//import breeze.linalg.{max, sum}
import java.math.BigDecimal
import java.util

import com.kongming.kmdm.common.db.bean.DataSourceBean
import com.kongming.kmdm.model.KmDataColumnType
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.expressions.Literal

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import org.apache.spark.sql.functions.{col, column, row_number}
import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
import org.apache.spark.sql.expressions._

import scala.util.Try
//import org.apache.spark.sql.functions._
import org.apache.spark.sql.functions.{col, row_number}

/**
  * Created by liuwei on 2017/11/8.
  */
object RowToColumn2 {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("RowToColumnTest").setMaster("local[8]")
    val sc = new SparkContext(sparkConf)
    val spark = SparkSession.builder

    val ss = spark.getOrCreate();
    import ss.implicits._


    import ss.implicits._

    val df = ss.createDataFrame(Seq(
      ("张三", "女", "数学", 60, 15, "张老师"),
      ("张三", "女", "语文", 70, 12, "李老师"),
      ("张三", "女", "物理", 50, 22, "张老师"),
      ("王二", "男", "数学", 50, 16, "赵老师"),
      ("王二", "男", "语文", 60, 20, "孙老师"),
      ("王二", "男", "历史", 50, 22, "李老师")
    )).toDF("姓名", "性别", "科目", "成绩", "排名", "监考老师")
    import scala.collection.JavaConverters._
    import org.apache.spark.sql.functions._
    df.createOrReplaceTempView("test")
    val title = Seq("姓名", "性别", "科目", "成绩", "排名", "监考老师")
    val rowLabel = Seq("姓名")
    val columnLabel = List("科目", "监考老师")
    val valueLabel = Seq("成绩", "排名")
    //    val keyArr = ArrayBuffer.empty[String]
    val keyColumnArr = df.select(columnLabel.head, columnLabel.tail: _*).distinct().collect().map(_.get(0).toString)
//    val columnValueArr = columnLabel.map(column => {
//      val value = df.select(column).distinct().rdd.collect().map(_.get(0).toString)
//      (column, value)
//    }
//    )

    var total = 1
    val columnValueMap = columnLabel.map(column => {
      val value = df.select(column).distinct().rdd.collect().map(_.get(0).toString)
      if (!value.isEmpty) {
        total = total * value.length
      }
      (column, value)
    }
    ).toMap

    val arr = ArrayBuffer.empty[String]
    rowLabel.foreach(arr += _)
    //    arr += rowLabel:_*
    //    val columnArr = ArrayBuffer.empty[String]
    val keyArr = arr.clone()
    columnLabel.foreach(keyArr += _)
    //    val keyRowArr = df.select(rowLabel).distinct().collect().map(_.get(0).toString)

    for (keyColumn <- keyColumnArr)
      for (value <- valueLabel) {
        val sb = new StringBuilder()
        sb.append(columnLabel(0))
        sb.append("(")
        sb.append(keyColumn)
        sb.append(")_value(")
        sb.append(value)
        sb.append(")")
        arr += sb.toString()
      }

    //    if(df.select(keyArr.head,keyArr.tail:_*).distinct().count() < df.count()) throw new Exception("数据行列标签不能确定唯一值！")


    val groupBy = df.rdd.groupBy(row => for (rowLable <- rowLabel) yield row.getString(title.indexOf(rowLable)))
    val qqq = groupBy.count()


    val titleNames = df.schema.fieldNames
    // 生成列标签的排列组合以及Column-SQL
    val a = ArrayBuffer.empty[KmDataColumnType]
    a += KmDataColumnType.CATEGORY
    a += KmDataColumnType.CATEGORY
    //    a += KmDataColumnType.CATEGORY
    val columnTagCombinedResult = wholeCombination(df, columnLabel, a.toList)
    println(columnTagCombinedResult)
    val rdd = groupBy.map(line => {
      println(line._1)
      println(line._2)

      var res = new ArrayBuffer[Any]
      res += line._1.head
      import scala.util.control.Breaks._
      for (columnKeyValue <- columnTagCombinedResult) {
        var flag = true
        breakable {
          line._2.foreach(row => {
            flag = true
            for (keyValue <- columnKeyValue) {
              val flag2 = if (row.get(title.indexOf(keyValue._1)).equals(keyValue._2)) true else false
              flag = flag & flag2
            }

            if (flag) {
              res += row.get(3) + ""
              res += row.get(4) + ""
              println(row.get(3) + "===" + row.get(4))
              break
            }
          })
        }
        if (!flag) {
          res +=  ""
          res +=  ""
          println("===")
        }
      }
      Row.fromSeq(res)
    })
    /* val result = ArrayBuffer.empty[List[(String, String)]]
     (0 until total).foreach(i => {
       var index = i
       val rowList = ArrayBuffer.empty[(String, String)]
       for (j <- columnLabel.indices) {
         val columnName = columnLabel(j)
         var mod = 1
         for (k <- j + 1 until columnLabel.size) {
           mod = mod * columnValueMap(columnLabel(k)).size
         }
         val valueIndex = index / mod
         index = index % mod
         rowList += columnName -> columnValueMap(columnName)(valueIndex)
       }
       result += rowList.toList
     })
     println(result)*/


    /*  val rdd = groupBy.map(line => {
        println(line._1)
        println(line._2)
        val rows = line._2
        var res = new ArrayBuffer[Any]
        res += line._1.head
        println("==============")

        columnLabel.foreach(column => {
          val values = columnValueMap(column)
          values.foreach(value => {
            println("----" + value)
          }
          )
        })


        Row.fromSeq(res)
      })
      rdd.count()*/

    /*  val rdd = groupBy.map(line => {
        println(line._1)
        println(line._2)
        var res = new ArrayBuffer[Any]
        res += line._1.head
        for (columnValue0 <- columnValueArr(0)._2) {
          for (columnValue1 <- columnValueArr(1)._2) {
            //           val columnContext = columnValue._2
            println("---" + columnValue0 + "---" + columnValue1)
            var flag = false
            line._2.foreach(row => {
              if (columnValue0.equals(row.getString(2)) && columnValue1.equals(row.getString(5))) {
                println("----" + row.get(3) + "==" + row.get(4))
                val a = res.clone()
                valueLabel.foreach(value=>{
                  res += row.get(title.indexOf(value))+""
                })
                flag = true
              }
            })
            if (!flag) {
              println("----" + "" + "==" + "")
              res += ""
              res += ""
            }
          }
        }
        println("==============")
        Row.fromSeq(res)
      })*/
//      var schema = df.select(rowLabel.head, rowLabel.tail: _*).schema
    var schema: StructType = StructType(Seq())
      for (i <- columnTagCombinedResult) {
        for (j <- 0 until 2)
        schema = schema.add(StructField(i(0)._2.substring(0,1)+i(1)._2.substring(0,1) , DataTypes.StringType))
      }

//      val df2 = ss.createDataFrame(rdd, schema).show(false)
    //    println(9)


    //    res2.columns
    //    println(res2)

    //    println(map)
    //    df.collect().foreach(line=>{
    //      val key1 = line.getAs[String](rowLabel)
    //             val key2 = line.getAs[String](columnLabel)
    //             val value = (line.getAs[Int]("成绩"),line.getAs[Int]("排名"))
    //             val a=  Map(key2 ->value)
    //             if(!map.contains(key1)){
    //               map += (key1 -> a)
    //             }else{
    //               var valueMap = map.get(key1).get
    //               valueMap += (key2 ->value)
    //               map += (key1 -> valueMap)
    //             }
    //    })
    //    var resDF = df.select(rowLabel).distinct().rdd
    //    for(i <- 0 until keyColumnArr.size){
    //      resDF.zip(df.filter(_.getAs[String](columnLabel) == keyColumnArr(0)).rdd)
    //    }
    //    println("=============="+resDF.first())
    //    resDF.first().get(0)
    //    resDF.map(row=>{
    //
    //      for(i <- 0 until row.size){
    //        print(row.get(i))
    //      }
    //    })
    //    println(resDF.count())


    //    val df2 = ss.createDataFrame(map.toSeq).show
    //    val a = df
    //     df.foreach(line =>{
    //       val key1 = line.getAs[String](rowLabel)
    //       val key2 = line.getAs[String](columnLabel)
    //       val value = (line.getAs[String]("成绩"),line.getAs[String]("科目"))
    //       val a=  Map(key2 ->value)
    //       if(!map.contains(key1)){
    //         map += (key1 -> a)
    //       }else{
    //         var valueMap = map.get(key1).get
    //         valueMap += (key2 ->value)
    //       }
    //     })
    //    val a = df.groupByKey(row=>{
    //      row.
    //    })
    //    println(map)

    //    a.

    //    val w = Window.partitionBy(rowLabel).orderBy($"成绩".desc)
    //    df.select($"姓名",sum($"排名").over(w).as("sum_duration")).show
    // val sqlContext = new SQLContext(sc)
    //    sqlContext.sql(
    //        " select * from test ").show

    //    val dfTop3 = re.withColumn("rank", row_number.over(w)).where($"rank" <= 3).show

    //    val a = df.groupBy(rowLabel).agg()

    //    println(a.map)

    //     df.groupBy(rowLabel).pivot(columnLabel).max("成绩","排名").show(false)
    //    df.rdd.aggregateByKey(0)(math.max(_, _), _ + _).collect.foreach(print)
    //    val tuples = df.rdd.aggregate(List.empty[List[String]])(seqOp = (result, lineWithIndex) => {
    //         List(line)
    //    }, combOp = (x, y) => x ::: y)
    //    tuples.foreach(println(_))

    //    val res2 = df.groupBy(rowLabel).pivot(columnLabel).max("成绩","排名").show(false)
    ////    val res = documentDF.groupBy(rowLabel).pivot(columnLabel,calueLabel).agg()
    //    val res= df.groupBy(rowLabel).pivot(columnLabel)
    //      .agg(Map(
    //      "成绩" -> "all",
    //      "排名" -> "max"
    //     ))
    //    res.show(false)
  }


  private def wholeCombination(df: DataFrame, columnTags: List[String], columnTagTypes: List[KmDataColumnType]): List[List[(String, String)]] = {
    val result = ArrayBuffer.empty[List[(String, String)]]

    var total = 1
    val columnTagValues = scala.collection.mutable.HashMap.empty[String, List[String]]
    columnTags.indices.foreach(index => {
      val titleName = columnTags(index)
      val titleType = columnTagTypes(index)
      columnTagValues += titleName -> {
        val values = df.select(titleName).distinct.collect
        if (values.isEmpty) List("")
        else {
          total = total * values.length
          values
            .map(
              row =>
                titleType match {
                  case KmDataColumnType.DOUBLE =>
                    if (row.isNullAt(0)) "" else Try(new BigDecimal(row.getDouble(0)).stripTrailingZeros.toPlainString).getOrElse("")
                  case KmDataColumnType.DATETIME => if (row.isNullAt(0)) "" else Try(row.getLong(0).toString).getOrElse("")
                  case _ => if (row.isNullAt(0)) "" else Try(row.getString(0)).getOrElse("")
                }
            )
            .toList
        }
      }
    })

    (0 until total).foreach(i => {
      var index = i
      val rowList = ArrayBuffer.empty[(String, String)]
      for (j <- columnTags.indices) {
        val columnName = columnTags(j)
        var mod = 1
        for (k <- j + 1 until columnTags.size) {
          mod = mod * columnTagValues(columnTags(k)).size
        }
        val valueIndex = index / mod
        index = index % mod
        rowList += columnName -> columnTagValues(columnName)(valueIndex)
      }
      result += rowList.toList
    })

    //    log.info("combineColumnTagValues=" + result.toList.toString())
    result.toList
  }

  private def getTitleTypes(dataSourceBean: DataSourceBean, titles: Seq[String]): List[KmDataColumnType] = {
    if (titles.isEmpty) return List.empty[KmDataColumnType]

    val titleNames = dataSourceBean.getTitleNames
    val titleTypes = dataSourceBean.getTitleTypes
    val result = ArrayBuffer.empty[KmDataColumnType]
    titles.foreach(titleName => result += titleTypes.get(titleNames.indexOf(titleName)))
    result.toList
  }
}
